package mayor.mq.quickstart;

import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;

import java.util.List;

public class Consumer {
    public static void main(String[] args) throws Exception{
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("quickstart-consumer");


        consumer.setNamesrvAddr("192.168.2.246:9876;192.168.2.190:9876");
        /**
         * 设置consumer第一次自动从队列头部开始消费还是队列尾部开始消费
         * 如果非第一次启动，那么按照上次消费的位置继续消费
         */

        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        /**
         *   * 号代表子主题过滤用
         */
        consumer.subscribe("TopicQuickStart","*");

        consumer.setConsumeMessageBatchMaxSize(10);

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                System.out.println(Thread.currentThread().getName()+ "Receive New Message: " + msgs);
                System.out.println("消息条数" + msgs.size());
                MessageExt message = msgs.get(0);
                try{
                    /**
                    for(MessageExt msg : msgs) {
                        String topic = msg.getTopic();
                        String msgBody = new String(msg.getBody(),"utf-8");
                        String tags = msg.getTags();
                        System.out.println("收到的消息 "+  " topic : " + topic + ",tags:" + tags + ",msg:"+msgBody);
                        if("Hello RocketMQ 4".equals(msgBody)){
                            System.out.println("========消息开始=======");
                            System.out.println(msg);
                            System.out.println("========消息结束=======");
                           int a = 1/0;
                        }
                    }
                    **/

                    String topic = message.getTopic();
                    String msgBody = new String(message.getBody(),"utf-8");
                    String tags = message.getTags();
                    System.out.println("收到的消息 "+  " topic : " + topic + ",tags:" + tags + ",msg:"+msgBody);
                    /**
                    if("Hello RocketMQ 4".equals(msgBody)){
                        System.out.println("========消息开始=======");
                        System.out.println(message);
                        System.out.println("========消息结束=======");
                        int a = 1/0;
                    }
                     **/
                }catch (Exception e){
                    if(message.getReconsumeTimes() == 3) {
                        //记录日志操作 重试后扔没发送，那就默认成功放到日志里
                        System.out.println("重试2次以后，还是没成功记录日志做处理");
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }


                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.out.println("Consumer Started. ");
    }
}
